package com.lagou.rpc.api.impl;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.map.MapUtil;
import com.lagou.rpc.api.NodeChangeListener;
import com.lagou.rpc.api.RpcRegistryCenter;
import com.lagou.rpc.common.RequestMetrics;
import com.lagou.rpc.config.ZookeeperConfig;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * @param
 * @author oyz
 * @version 1.0
 * @description: TODO 注册中心接口类的实现类
 * @date 2022/2/27 18:09
 */
public class ZookeeperRegistryCenter implements RpcRegistryCenter {


    private CuratorFramework client;

    private static final String LG_EDU_RPC_ZK_ROOT = "/lg-rpc-provider";

    /**
     * 快捷获取从线程池拿去线程 获取一个核心线程拿去跑业务(无界队列,任务数过多，有可能会产生OOM的问题)
      */
    private static final ScheduledExecutorService REPORT_WORKER = Executors.newScheduledThreadPool(1);

    /**
     * 存储子节点数组
     */
    private List<String> serviceList = new CopyOnWriteArrayList<>();
    /**
     * 存放变更的节点数组
     */
    private static List<NodeChangeListener> listenerList = new ArrayList<>();

    /**
     *定义一个全局的zk的server地址
     */
    private String connectString;



    /**
     * @description: TODO 完成注册节点和写入日志的操作
     * @author oyz
     * @param connectString 连接zk的server地址,如:   server:2181
     * @date 2022/2/27 18:17
     * @version 1.0
     */
    public ZookeeperRegistryCenter(String connectString){
        this.connectString = connectString;
        // 定义连接和会话的超时时间,由于我的机器比较慢，我把它设置长一点
        int timeOut = 20000;
        // 使用curator客户端的api来连接Zookeeper
        client = CuratorFrameworkFactory.builder()
                .connectString(connectString)
                .sessionTimeoutMs(timeOut)
                .connectionTimeoutMs(timeOut)
                // 每次连接睡眠1秒,重试1次
                .retryPolicy(new RetryNTimes(1, 1000))
                .build();

        client.getConnectionStateListenable().addListener((CuratorFramework curatorFramework, ConnectionState connectionState) ->{
            if(ConnectionState.CONNECTED.equals(connectionState)){
                System.out.println("注册中心注册成功!");
            }
        });

        client.start();

        //todo 客户端需要每20秒上报一次(因为我的电脑连接我的VM比较慢)
        //判断它是否是客户请求
        if(ZookeeperConfig.getInstance().isConsumerSide()&&ZookeeperConfig.getInstance().getReportTime()>0) {
            // 用线程来执行下面的逻辑
            REPORT_WORKER.scheduleWithFixedDelay(() -> {
                ConcurrentHashMap<String, Long> metriesMap = RequestMetrics.getInstance().getMetriesMap();
                if (MapUtil.isEmpty(metriesMap)) {
                    return;
                }
                System.out.println("自动上报节点耗时日志:" + metriesMap);
                metriesMap.entrySet().forEach(entry -> {
                    // 如: 127.0.0.1:2181
                    String address = entry.getKey();
                    // 花费时间
                    Long cost = entry.getValue();
                    // 先创建路径 /lg-rpc-provider/metrics
                    String path = LG_EDU_RPC_ZK_ROOT + "/metrics";
                    // 不存在该节点
                    if (!exists(path)) {
                        // 创建节点
                        create(path, false);
                    }
                    String instancePath = path + "/" + address;
                    // 更新节点信息
                    updateWithData(instancePath, cost.toString(), true);

                });

            },ZookeeperConfig.getInstance().getReportTime(),ZookeeperConfig.getInstance().getReportTime(), TimeUnit.SECONDS);

        }
    }

    /**
     * @description: TODO 更新节点的方法
     * @author oyz
     * @param
     * @date 2022/2/28 9:48
     * @version 1.0
     */
    private void updateWithData(String instancePath, String content, boolean b) {
        if (b) {
            if (exists(instancePath)) {
                update(instancePath, content);
            } else {
                this.createEphemeral(instancePath, content);
            }
        } else {
            if (exists(instancePath)) {
                update(instancePath, content);
            } else {
                this.createPersistent(instancePath, content);
            }
        }

    }

    /**
     * @description: TODO 更新节点的方法
     * @author oyz
     * @param path
     * @param data
     * @date 2022/2/28 10:04
     * @version 1.0
     */
    private void update(String path,String data){
        try {
            byte[] bytes = data.getBytes("utf-8");
            client.setData().forPath(path,bytes);
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * @description: TODO 根据传入的参数创建两种不同类型的节点
     * @author oyz
     * @param path
     * @param flag
     * @date 2022/2/27 21:56
     * @version 1.0
     */
    private void create(String path,boolean flag) {
        if(flag){
            // 创建临时节点
            this.createEphemeral(path,null);
        }else {
            // 创建持久性节点
            this.createPersistent(path,null);
        }
    }

    /**
     * @description: TODO 创建永久节点
     * @author oyz
     * @param path 节点路径
     * @param data 节点保存的信息
     * @date 2022/2/28 9:21
     * @version 1.0
     */
    private void createPersistent(String path,String data) {
        try {
            if(data == null){
                client.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path);
            }else{
                byte[] bytes = data.getBytes();
                client.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path,bytes);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * @description: TODO 创建临时节点
     * @author oyz
     * @param path 节点路径
     * @param data 节点保存的信息
     * @date 2022/2/28 9:18
     * @version 1.0
     */
    private void createEphemeral(String path,String data) {
        try {
            if(data == null){
                client.create().creatingParentContainersIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path);
            }else{
                byte[] bytes = data.getBytes();
                client.create().creatingParentContainersIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path,bytes);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    /**
     * 判断节点是否存在
     * @param path Zookeeper的节点路径
     * @return
     */
    private boolean exists(String path) {
        try {
            Stat stat = client.checkExists().forPath(path);
            if (stat != null){
                return true;
            }
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(),e);
        }
        return false;
    }



    @Override
    public boolean registry(String service, String ip, int port) {
        // /lg-edu-rpc/com.lagou.edu.api.IUserService/provider
        String zkPath = providerPath(service);
        // 判断该路径在zk上存不存在
        if(!exists(zkPath)){
            // 创建永久节点
            create(zkPath,false);
        }
        // /lg-edu-rpc/com.lagou.edu.api.IUserService/provider/127.0.0.1:8081
        String instancePath = zkPath + "/" + ip + ":" + port;
        // 创建临时节点
        create(instancePath,true);
        return true;
    }

    /**
     * @description: TODO 创建节点路径
     * @author oyz
     * @param
     * @date 2022/2/28 10:14
     * @version 1.0
     */
    private String providerPath(String service) {
        return LG_EDU_RPC_ZK_ROOT+service+"/provider";
    }

    /**
     * @description: TODO 发现服务
     * @author oyz
     * @param  service 节点名称
     * @date 2022/2/28 10:26
     * @version 1.0
     */
    @Override
    public List<String> discovery(String service) {
        // /lg-edu-rpc/com.lagou.rpc.api.IUserService/provider
        String path = providerPath(service);
        try {
            // 判断子节点列表是否为空,为空则需要从注册中心拉取下来
            if (CollectionUtil.isEmpty(serviceList)) {
                //  /provider
                // [127.0.0.1:7777,127.0.0.1:7778]
                serviceList = client.getChildren().forPath(path);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
        // 监听当前目录节点的信息变更
        this.registryWatch(service,path);
        return serviceList;
    }

    /**
     * @description: TODO 监听节点信息变更的方法
     * @author oyz
     * @param service 节点名称   /provider
     * @param path 节点路径   /lg-edu-rpc/com.lagou.edu.api.UserService/provider
     * @date 2022/2/28 10:51
     * @version 1.0
     */
    // 用final修饰参数的原因是为了防止该地址在我方法内部被篡改
    private void registryWatch(final String service,final String path) {
        PathChildrenCache nodeCache = new PathChildrenCache(client,path,true);
        try {
            nodeCache.getListenable().addListener((curatorFramework,pathChildrenCacheEvent)->{
                // 更新本地缓存
                serviceList = client.getChildren().forPath(path);
                // 一旦发生变化,通知到每一个子节点
                listenerList.stream().forEach(nodeChangeList ->{
                    System.out.println("节点变化,通知每一个子节点");
                    nodeChangeList.notify(service,serviceList,pathChildrenCacheEvent);
                });
            });
            // /lg-edu-rpc/com.lagou.edu.api.UserService/provider/127.0.0.1:8990
            /**
             * StartMode：初始化方式
             * POST_INITIALIZED_EVENT：异步初始化。初始化后会触发事件
             * NORMAL：异步初始化
             * BUILD_INITIAL_CACHE：同步初始化
             * */
            nodeCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
        }catch (Exception e){
            e.printStackTrace();
        }

    }

    @Override
    public void addListener(NodeChangeListener nodeChangeListener) {
        listenerList.add(nodeChangeListener);
    }

    @Override
    public void destroy() {
        if (null!= client){
            client.close();
        }
    }
}
